package com.ww.flink

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 需求：
 * 每隔一段时间对最近一段时间内，所有的车辆信息按照速度排序 并且获取最大最小速度的车辆信息
 *
 * 排序：全量聚合函数
 */
object Flink_try14_all_agg {
  case class CarInfo(carId: String, speed: Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val initStream: DataStream[String] = env.socketTextStream("node01", 8888)
    initStream.map(data =>{
      val splits = data.split(" ")
      //车牌号   车速
      (splits(0),splits(0).toLong)
    }).timeWindowAll(Time.minutes(30),Time.seconds(5))
      .process(new ProcessAllWindowFunction[(String,Long),Iterable[(String, Long)],TimeWindow] {
        /**
         * @param context  上下文对象
         * @param elements  窗口中所有的元素
         * @param out      往下游发射数据的对象
         */
        override def process(context: Context, elements: Iterable[(String, Long)], out: Collector[Iterable[(String, Long)]]): Unit = {
          //elements集合排序
          val sortList = elements.toList.sortBy(_._2)
          val maxSpeedCarInfo = sortList.last._1
          val minSpeedCarInfo = sortList.head._1
          println("最大车速的车牌号：" + maxSpeedCarInfo + "\t最小车速的车牌号：" + minSpeedCarInfo)
          out.collect(sortList)
        }
      }).print()





    env.execute()


  }


}
